package com.imooc.bigdata.chapter08_my.business

import com.imooc.bigdata.chapter08.utils.{KuduUtils, SQLUtils, SchemaUtils}
import com.imooc.bigdata.chapter08_my.utils.IPUtils
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SparkSession}

/*
    主要是需求分析中的ETL操作，把原始数据进行ETL，存储到kudu ods表中，供后续使用
 */
object LogETLApp {

  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession
      .builder().appName("LogETLApp")
      .master("local[2]")
      .getOrCreate()

    import spark.implicits._
    //加载要处理的数据源(Data Source)，这里是本地
    var jsonDF: DataFrame = spark.read.json("E:\\IntelliJ-IDEA\\sparksql-train\\data\\data-test.json")

    /*
       1  这里是网上的ip地址库，然后找出对应的地区
       2  也可以用gitHub 上面的 纯真的ip解析
       3  但是生产的时候，用专门的公司，付钱的ip服务
     */
    //加载本地ip库
    val ipRDD: RDD[String] = spark.sparkContext
      .textFile("E:\\IntelliJ-IDEA\\sparksql-train\\data\\ip.txt")

    //把这个RDD读取的数据拆分出需要的字段，然后转成DF，注册成一张规则表
    val ipRuleDF: DataFrame = ipRDD.map(x => {
      var splits: Array[String] = x.split("\\|")
      val startIP: Long = splits(2).toLong
      val endIP: Long = splits(3).toLong
      val province: String = splits(6)
      val city: String = splits(7)
      val isp: String = splits(9)
      (startIP, endIP, province, city, isp)
    }).toDF("start_ip", "end_ip", "province", "city", "isp")

    /*
     将数据中的ip和ipRuleDF中的ip对应，得到相应的省份，城市，网络运营商。
     现在有两个DF，一个是原始数据DF 一个是规则化DF, 合并用join操作， 条件是ipRdd的ip得在ipRuleDF ip 范围内

    但是json里面的ip是字符串，192.168.2.19  ip地址库里面的是3059465949  Long类型
    所以下面需要转换需要转换
     */

    //自定义函数， 调用ip转换类
    import org.apache.spark.sql.functions._
    def getLongIP() = udf((ip:String)=>{
      IPUtils.ip2Long(ip)
    })
    jsonDF = jsonDF.withColumn("ip_Long",getLongIP()($"ip"))

//    //API方式
//    jsonDF.join(ipRuleDF,jsonDF("ip_Long")
//      .between(ipRuleDF("start_ip"),ipRuleDF("end_ip")))
//        .show()

      //SQL方式
    jsonDF.createOrReplaceTempView("logs")
    ipRuleDF.createOrReplaceTempView("ips")

    val sql = SQLUtils.SQL
    val dataFrame: DataFrame = spark.sql(sql)
    val MASTER = "spark000"
    val TABLE_NAME = "ods"
    val partitionID = "ip"
    val schema = SchemaUtils.ODSSchema

    KuduUtils.sink(dataFrame,TABLE_NAME,MASTER,schema,partitionID)

    spark.stop()
  }
}
